package com.mjf.day2

import org.apache.flink.api.common.functions.{FilterFunction, FlatMapFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object FlatMapExample {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream: DataStream[String] = env.fromElements("white", "gray", "black")

    // 针对流中每一个元素，生成0个，1个，或多个数据
    stream.flatMap(new MyFlatMapFunction()).print()

    val sensorStream: DataStream[SensorReading] = env.addSource(new SensorSource)

    // 使用flatMap实现map功能
    sensorStream.flatMap(new FlatMapFunction[SensorReading, String] {
      override def flatMap(t: SensorReading, collector: Collector[String]): Unit = {
        collector.collect(t.id)
      }
    }).print()

    // 使用flatMap实现filter功能
    sensorStream.flatMap(new FlatMapFunction[SensorReading, SensorReading] {
      override def flatMap(t: SensorReading, collector: Collector[SensorReading]): Unit = {
        if (t.id.equals("sensor_1")) {
          collector.collect(t)
        }
      }
    }).print()

    env.execute("FlatMapExample")

  }

  class MyFlatMapFunction extends FlatMapFunction[String, String] {
    // 使用`out.collect()`方法，将数据发往下游
    override def flatMap(value: String, out: Collector[String]): Unit = {
      if(value.equals("white")) {
        out.collect(value)
      } else if(value.equals("black")) {
        out.collect(value)
        out.collect(value)
      }
    }
  }

}
